<?php
// +----------------------------------------------------------------------
// | Workman and Gateway-worker
// +----------------------------------------------------------------------
// | Copyright (c) 2015-2025 http://www.yicmf.com, All rights reserved.
// +----------------------------------------------------------------------
// | Author: 微尘 <yicmf@qq.com>
// +----------------------------------------------------------------------

namespace yicmf\worker\command;

use GatewayWorker\BusinessWorker;
use GatewayWorker\Gateway;
use GatewayWorker\Register;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use think\facade\App;
use think\facade\Config;
use Workerman\Timer;
use yicmf\worker\Http as HttpServer;
use yicmf\worker\Server as WorkerServer;
use yicmf\worker\Plan as WorkerPlan;
use Workerman\Worker as WorkermanWorker;

/**
 * Worker 命令行类
 */
class Worker extends Command
{
    protected $config = [];

    public function configure()
    {

        $this->setName('worker')
            ->addArgument('action', Argument::OPTIONAL, "start|stop|restart|reload|status|connections", 'start')
            ->addOption('step', 's', Option::VALUE_OPTIONAL, 'register|business|gateway the step of Gateway-worker ', 'register')
//            ->addOption('step', 's', Option::VALUE_OPTIONAL, 'the step of workerman server.', null) 
            ->setDescription('Workman and Gateway-worker');
    }

    public function execute(Input $input, Output $output)
    {
        $action = $input->getArgument('action');

        if (!in_array($action, ['start', 'stop', 'reload', 'restart', 'status', 'connections'])) {
            $output->writeln("Invalid argument action:{$action}, Expected start|stop|restart|reload|status|connections .");
            exit(1);
        }
        global $argv;
        array_shift($argv);
        array_shift($argv);
        array_unshift($argv, 'think', $action);

        $config = Config::get('worker');
        //热更新

        if ($config['hot_update']['enable']) {
            $this->addHotUpdateWorker($config['hot_update']);
        }
        if ($config['server']['enable']) {
            //启动server
            $this->_startServer($action, $config['server']);
        }
        if ($config['plan']['enable']) {
            //启动server
            $this->_startPlan($action, $config['plan']);
        }
        $step = $input->getOption('step');
        //gateway
        if ($config['gateway']['enable']) {
            $this->start($action, $config['gateway'], $step);
        }
        // 运行所有服务
        WorkermanWorker::runAll();

    }


    protected function _startServer($action, $option)
    {
        if ('start' == $action) {
            $this->output->writeln('Starting Workerman server...');
        }

        // 自定义服务器入口类
        if (!empty($option['handler'])) {
            $class = (array)$option['handler'];

            foreach ($class as $server) {
                $this->startServer($server, $option);
            }
        } else {
            $this->output->writeln('Starting err no handler');
        }

    }

    protected function _startPlan($action, $option)
    {
        if ('start' == $action) {
            $this->output->writeln('Starting Workerman server...');
        }

        // 自定义服务器入口类
        if (!empty($option['handler'])) {
            $class = (array)$option['handler'];
            foreach ($class as $server) {
                $this->startPlan($server, $option);
            }
        } else {
            $this->output->writeln('Starting err no handler');
        }

    }

    protected function startServer(string $class, $option)
    {
        if (class_exists($class)) {
            $worker = new $class($option);
            if (!$worker instanceof WorkerServer) {
                $this->output->writeln("<error>Worker Server Class Must extends \\yicmf\\worker\\Server</error>");
            }
        } else {
            $this->output->writeln("<error>Worker Server Class Not Exists : {$class}</error>");
        }
    }

    protected function startPlan(string $class, $option)
    {
        if (class_exists($class)) {
            $worker = new $class($option);
            if (!$worker instanceof WorkerPlan) {
                $this->output->writeln("<error>Worker Plan Class Must extends \\yicmf\\worker\\Plan</error>");
            }
        } else {
            $this->output->writeln("<error>Worker Plan Class Not Exists : {$class}</error>");
        }
    }


    protected function addHotUpdateWorker($option)
    {
        if (DIRECTORY_SEPARATOR !== '\\') {
            $worker = new \Workerman\Worker();
            ///暂时支持Linux
            $worker->name = 'hot update';
            $worker->reloadable = false;
            $worker->onWorkerStart = function ()use($option) {
                Timer::add(1, function ()use($option)  {
                    global $last_mtime;
                    foreach ($option['include'] as $item) {
                        // recursive traversal directory
                        $dir_iterator = new \RecursiveDirectoryIterator($item);
                        $iterator = new \RecursiveIteratorIterator($dir_iterator);
                        foreach ($iterator as $file) {
                            // only check php files pathinfo($file, PATHINFO_EXTENSION)
                            if (!in_array(pathinfo($file, PATHINFO_EXTENSION),$option['ext'])) {
                                continue;
                            }
                            // check mtime
                            if ($last_mtime < $file->getMTime()) {
                                echo $file . " update and reload\n";
                                // send SIGUSR1 signal to master process for reload
                                posix_kill(posix_getppid(), SIGUSR1);
                                $last_mtime = $file->getMTime();
                                break;
                            }
                        }
                    }
                });
            };
        }
    }

    /**
     * 启动
     * @access public
     * @param string $host 监听地址
     * @param integer $port 监听端口
     * @param array $option 参数
     * @return void
     */
    public function start(string $action, array $option = [], $step = 'register')
    {
        $registerAddress = !empty($option['registerAddress']) ? $option['registerAddress'] : '127.0.0.1:1236';

        if (DIRECTORY_SEPARATOR === '\\') {

            if ($step == 'register') {
                $this->register($registerAddress);
            }
            if ($step == 'business') {
                $this->businessWorker($registerAddress, $option);
            }
            if ($step == 'gateway') {
                $this->gateway($registerAddress, $option);
            }
        } else {
            if (!empty($option['register']['deploy'])) {
                // 分布式部署的时候其它服务器可以关闭register服务
                // 注意需要设置不同的lanIp
                $this->register($registerAddress);
            }
            // 启动businessWorker
            $this->businessWorker($registerAddress, $option);
            // 启动gateway
            $this->gateway($registerAddress, $option);
        }
    }

    /**
     * 启动register
     * @access public
     * @param string $registerAddress
     * @return void
     */
    public function register(string $registerAddress)
    {
// register 必须是text协议，切记不能将register端口开放给外网
        new Register('text://' . $registerAddress);
    }

    /**
     * 启动businessWorker
     * @access public
     * @param string $registerAddress registerAddress
     * @param array $option 参数
     * @return void
     */
    public function businessWorker(string $registerAddress, array $option = [])
    {
        // 初始化 bussinessWorker 进程
        $worker = new BusinessWorker();

//        $this->option($worker, $option);

// 服务注册地址
        $worker->registerAddress = $registerAddress;
        $worker->eventHandler = $option['business']['handler'];
// bussinessWorker 进程
        $worker->name = $option['business']['name'];
        // bussinessWorker进程数量
        $worker->count = $option['business']['count'];
        BusinessWorker::$pidFile = App::getRootPath() . 'runtime/business_worker.pid';
        BusinessWorker::$logFile = App::getRootPath() . 'runtime/business_worker.log';


    }

    /**
     * 启动gateway
     * @access public
     * @param string $registerAddress registerAddress
     * @param string $host 服务地址
     * @param integer $port 监听端口
     * @param array $option 参数
     * @return void
     */
    public function gateway(string $registerAddress, array $option = [])
    {

        // 初始化 gateway 进程
        if (!empty($option['socket'])) {
            $socket = $option['socket'];
            unset($option['socket']);
        } else {
            $socket = $option['protocol'] . '://' . $option['host'] . ':' . $option['port'];
            unset($option['host'], $option['port'], $option['protocol']);
        }

//        $gateway = new Gateway($socket, $option['context'] ?? []);
        $gateway = new Gateway($socket);

        // 以下设置参数都可以在配置文件中重新定义覆盖
        $gateway->name = $option['gateway']['name'];
// gateway进程数，一般设置2个就足够
        $gateway->count = $option['gateway']['count'];
        // 本机ip，分布式部署时使用内网ip
        $gateway->lanIp = '127.0.0.1';
// 内部通讯起始端口，假如$gateway->count=2，起始端口为2900
// 则一般会使用2900 2901 2个端口作为内部通讯端口
        $gateway->startPort = 2900;
        $gateway->pingNotResponseLimit = $option['gateway']['pingNotResponseLimit'];
// 心跳间隔
        $gateway->pingInterval = $option['gateway']['pingInterval'];

        // 心跳数据
        $gateway->pingData =  $option['gateway']['pingData'];
// 服务注册地址
        $gateway->registerAddress = $registerAddress;

        // 全局静态属性设置
//        foreach ($option as $name => $val) {
//            if (in_array($name, ['stdoutFile', 'daemonize', 'pidFile', 'logFile'])) {
//                WorkermanWorker::${$name} = $val;
//                unset($option[$name]);
//            }
//        }
//
//        $this->option($gateway, $option);
    }

    /**
     * 设置参数
     * @access protected
     * @param WorkermanWorker $worker Worker对象
     * @param array $option 参数
     * @return void
     */
    protected function option(WorkermanWorker $worker, array $option = [])
    {
        // 设置参数
        if (!empty($option)) {
            foreach ($option as $key => $val) {
                $worker->$key = $val;
            }
        }
    }
}
